package com.xahj.bd2104.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * Created with IntelliJ IDEA.
 * Author: Amos
 * E-mail: amos@amoscloud.com
 * Date: 2021/9/14
 * Time: 10:34
 * Description: 
 */
object SparkCoreDemo7_transformations {
  def main(args: Array[String]): Unit = {
    val sc = new SparkContext(new SparkConf()
      .setAppName("demo7")
      .setMaster("local[2]"))

    // 1. cartesian
    // 笛卡尔积
    //def cartesian[U: ClassTag](other: RDD[U]): RDD[(T, U)] =
    val rdd = sc.makeRDD(List("米饭", "面条"))
    val rdd1 = sc.makeRDD(List("蘑菇炒肉", "土豆牛肉", "宫保鸡丁"))

    val rdd2 = rdd.cartesian(rdd1)
    //    rdd2.foreach(println)

    val rdd3 = sc.makeRDD(List("西安", "北京", "上海", "深圳", "武汉"))
    rdd3.cartesian(rdd3)
      .filter(t => t._1 != t._2)
    //      .foreach(println)

    // 2. 重新分区有两个算子
    // 2.1 repartition  会开启shuffle
    //   优点: 调整分区数量的同时 会将分区的数据进行均衡 避免数据倾斜
    //   缺点: 造成shuffle的IO
    //   使用场景 : 扩充分区个数
    // 2.2 coalesce 默认关闭shuffle
    //   优点: 不开启shuffle可以避免网络IO
    //   缺点: 无法改善数据倾斜
    //   使用场景 : 缩减分区个数

    val rdd4 = sc.makeRDD(1 to 100)
    val rdd5 = rdd4.repartition(10)
    rdd5.coalesce(5)

    // 3. 自定义分区
    // 自定义分区时， 需要使用kv泛型的RDD
    // 分区条件 按照Key进行分区
    val rdd6 = sc.makeRDD(1 to 10)
    val rdd7: RDD[(Int, Int)] = rdd6.map(x => (x, x))
    //    def partitionBy(partitioner: Partitioner): RDD[(K, V)] =
    //   重新分区时需要指定分区器
    // abstract class Partitioner extends Serializable {
    //  def numPartitions: Int
    //  def getPartition(key: Any): Int
    //}
    //  spark默认提供了两种分区器的实现
    //    HashPartitioner 按照Key的hashCode与给定的分区数取模的结果作为分区编号
    //        优点： 可以保证相同key的数据分到相同的分区
    //        缺点:  会造成数据倾斜
    //    RangePartitioner 将所有数据集使用抽样算法,抽取一个分配一个
    //        优点:  将数据均衡的分配到所有分区
    //        缺点:  相同Key的数据被散列到所有分区，不利于按照Key的聚合操作

    //    如果业务逻辑与上述的分区方式不符可以自定义分区实现分区逻辑
    //
    val rdd8 = rdd7.partitionBy(new ModPartitioner(3))
      .map(_._2)

    rdd8.mapPartitionsWithIndex((i, iter) => {
      Iterator(i + "-" + iter.mkString(","))
    })
    //      .foreach(println)

    // 4. 分区同时排序

    // 需要产生两个新的RDD
    // 第一次 仅仅进行分区
    // 第二次 对分区内的数据进行排序
    rdd7.partitionBy(new ModPartitioner(3))
      .sortByKey()

    // 只需要产生一个RDD
    // 在分区的同时将排好序的数据 写入新的RDD
    rdd7.repartitionAndSortWithinPartitions(new ModPartitioner(3))

    // 5. pipe

    // pipe接收一个bash命令或者shell脚本
    // pipe会将源RDD中的所有元素  依次传入脚本作为脚本的输入
    //  并将脚本的输出以String对象 接收并封装在新的RDD返回

    // 早期有一些使用python、shell、c编写的运维脚本或者程序
    // 只能在linux运行， 如果需要使用spark执行相同的逻辑
    //  两种解决方案: 1. 使用java或者scala将程序重写
    //              2. 使用pipe作为桥梁
    //                  将RDD的数据传入shell脚本
    //                  将shell脚本的计算结果接收到新的RDD

    val rdd9 = sc.makeRDD(1 to 10)
    //def pipe(command: String): RDD[String] =
    val rdd10 = rdd9.pipe("ls")
  }
}

